package com.strato.queue.v1_0_0.exec;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.strato.base.env.v1_0_0.GlobalEnvironment;
import com.strato.base.osgi.v1_0_0.BundleUtil;
import com.strato.batch.base.v1_0_0.api.Job;
import com.strato.batch.base.v1_0_0.api.JobProducer;
import com.strato.batch.base.v1_0_0.common.JobImpl;
import com.strato.jdbc.datasource.v1_0_0.DataSourceFactory;
import com.strato.orm.mybatis.v1_0_0.factory.MapperFactory;
import com.strato.queue.v1_0_0.activator.Activator;
import com.strato.queue.v1_0_0.api.QueueService;
import com.strato.queue.v1_0_0.jdbc.dao.mapper.MerchantClientapiNodeMapper;
import com.strato.queue.v1_0_0.jdbc.model.MerchantClientapiNode;
import com.strato.queue.v1_0_0.jdbc.model.MerchantClientapiNodeExample;
import com.strato.queue.v1_0_0.jdbc.model.QueueMessage;
import com.strato.saas.merchant.v1_0_0.MerchantContextHolder;
import com.strato.transaction.v1_0_0.api.jdbc.DataSourceTransactionManager;
import com.strato.transaction.v1_0_0.api.jdbc.Execution;

public class RoundRobinQueueJobProducer implements JobProducer<QueueMessage> {

	private String messageType;
	
	private boolean available=true;
	
	private int holdMilliSeconds=10*1000;
	
	private Map<String,Integer> ignoreMap=new HashMap<String,Integer>();
	
	public RoundRobinQueueJobProducer(String messageType,int holdMilliSeconds) {
		this.messageType=messageType;
		this.holdMilliSeconds=holdMilliSeconds;
	}

	@Override
	public Job<QueueMessage> produce() {
		return doProduce(1);
	}

	private Job<QueueMessage> doProduce(int n) {
		if(n<0){
			return null;
		}
		String merchantId=nextMerchant();
		Integer ignoreTimes=ignoreMap.get(merchantId);
		while(ignoreTimes!=null&&ignoreTimes>0){
			if(ignoreTimes>1){
				ignoreMap.put(merchantId, ignoreTimes-1);
			}else{
				ignoreMap.remove(merchantId);
			}
			if(_merchantIdIter.hasNext()){//try next
				merchantId=nextMerchant();
				ignoreTimes=ignoreMap.get(merchantId);
			}else{
				return doProduce(-1);
			}
		}
		final String finalMerchantId=merchantId;
		QueueMessage message=DataSourceTransactionManager.requiresNew(new Execution<QueueMessage>() {
			@Override
			public QueueMessage execute() {
				String backupMerchant=MerchantContextHolder.get();
				MerchantContextHolder.set(finalMerchantId);
				try{
					QueueMessage msg = getQueueService().receiveMessage(messageType, holdMilliSeconds);
					return msg;
				}finally{
					MerchantContextHolder.set(backupMerchant);
				}
			}
		});
		if(message==null){
			ignoreTimes=1;
			ignoreMap.put(merchantId, ignoreTimes);
			return doProduce(n-1);
		}
		JobImpl<QueueMessage> job=new JobImpl<QueueMessage>();
		job.setObject(message);
		return job;
	}

	private Iterator<String> _merchantIdIter=null;
	
	private synchronized String nextMerchant() {
		if(_merchantIdIter==null||!_merchantIdIter.hasNext()){
			String nodeId=GlobalEnvironment.getInstance().resolveString("${strato.queue.nodeId}");
			List<String> merchantIds=new ArrayList<String>();
			MerchantClientapiNodeExample example=new MerchantClientapiNodeExample();
			example.createCriteria().andNodeIdEqualTo(nodeId);
			MerchantClientapiNodeMapper merchantClientapiNodeMapper=MapperFactory.getMapper(MerchantClientapiNodeMapper.class, DataSourceFactory.COMMON_DATASOURCE);
			List<MerchantClientapiNode> merchants = merchantClientapiNodeMapper.selectByExample(example);
			if(merchants.isEmpty()){
				throw new NullPointerException("No merchants found for node:"+nodeId);
			}
			for(MerchantClientapiNode merchant:merchants){
				merchantIds.add(merchant.getMerchantId());
			}
			_merchantIdIter=merchantIds.iterator();
		}
		return _merchantIdIter.next();
	}

	@Override
	public boolean available() {
		return available;
	}

	public void setAvailable(boolean available) {
		this.available = available;
	}

	public QueueService getQueueService() {
		return BundleUtil.getCachedService(Activator.getContext(), QueueService.class, null);
	}

	public String getMessageType() {
		return messageType;
	}

	public void setMessageType(String messageType) {
		this.messageType = messageType;
	}

	public int getHoldMilliSeconds() {
		return holdMilliSeconds;
	}

	public void setHoldMilliSeconds(int holdMilliSeconds) {
		this.holdMilliSeconds = holdMilliSeconds;
	}
	
}
